fix: decode invalid UTF-8 at the JVM to native FFI import boundary - #5310

This PR continues the work from #4945.

Which issue does this PR close?

Part of #4764 (EPIC: consistent handling of invalid UTF-8 in native StringType). This PR implements Gap B only, decoding invalid UTF-8 at the JVM to native Arrow FFI import boundary. It does not close the EPIC, which still tracks Gap A (the native scan rejecting invalid UTF-8).

Rationale for this change

Spark's StringType (UTF8String) can hold arbitrary bytes, including sequences that are not valid UTF-8. When a JVM side source hands string columns to native code over the Arrow C Data Interface, arrow-rs imports them with from_ffi / from_ffi_and_data_type, which build the array via ArrayData::new_unchecked and do not validate UTF-8. The imported Arrow Utf8 / LargeUtf8 array then lies about its validity, and any downstream native string kernel that reads &str through arrow-rs's unchecked StringArray::value() (from_utf8_unchecked) exercises undefined behaviour: iterating chars, slicing on char boundaries, and similar operations can misbehave, panic, or be miscompiled. This is a latent, default configuration soundness hazard.

The string producing sites already decode invalid bytes the way Spark renders them: CAST(binary AS string) (#4763) and native shuffle get_string (#4521) both use decode_utf8_spark_lossy. This PR applies the same policy to the string ingress side, so the whole native pipeline agrees on a single invariant: native string data is always valid UTF-8.

What changes are included in this PR?

• A new decode_string_arrays walker in datafusion-comet-common, beside the existing decode_utf8_spark_lossy. It ensures every Utf8 / LargeUtf8 array reachable from an imported column holds valid UTF-8, decoding invalid bytes to Spark's rendered form. It is zero copy for the valid common case (one validation pass plus an O(number of strings) boundary check, returning the same Arc), and rebuilds element by element only when bytes are genuinely invalid. It recurses through Dictionary, Struct, List, LargeList, FixedSizeList, and Map.

• There are four from_ffi call sites. Three carry string data and decode directly:

  • ScanExec::pull_next, which handles all native query input (the JVM reader, Spark columnar handoff, shuffle reads, mapInArrow). Decoding runs before dictionary unpack so compact dictionary values are validated rather than the expanded ones.
  • columnarToRow conversion.
  • JVM UDF result.

  The fourth, batch_from_ffi in aligned_stream_reader.rs, is consumed only by ScanExec, which passes every column through import_column immediately after the batch is materialised. Decoding there as well would validate the same buffers twice, so its doc comment records that any future consumer must keep that step.

• Anything string-bearing that the walker does not handle fails closed rather than passing unchecked data through. data_type_contains_string is an exhaustive match, so an Arrow upgrade that adds a data type forces this decision to be revisited, and types such as Utf8View, the view lists and RunEndEncoded return NotYetImplemented if they ever reach this boundary.

• The fast path validates the used byte range and also confirms no element boundary splits a codepoint. This second check is required for soundness: a whole buffer valid "é" (bytes C3 A9) split across two offsets would otherwise hand each element an invalid slice, and value() decodes those unchecked.

• Validation uses simdutf8 rather than std::str::from_utf8 (see Performance below). simdutf8 is already in the dependency graph through arrow-json, so this adds no new third-party crate.

• Benchmarks: a criterion benchmark for the valid fast path, and CometStringFfiImportBenchmark, which measures both import sites end to end.

No configuration flag gates this. It fixes undefined behaviour in the default configuration and matches the always on behaviour of the sibling cast and shuffle fixes.

The only observable divergence from Spark is the previously documented one: decoding rather than preserving raw bytes differs only under byte level round trips (for example CAST(CAST(X'FF' AS STRING) AS BINARY)), already noted in the compatibility guide.

Performance

Validation costs one pass over each imported string buffer, so the cost scales with the bytes crossing the boundary, not with the number of rows.

CometStringFfiImportBenchmark (added here) measures both import sites over 1M rows of roughly 180 byte strings, in ASCII and in multibyte text. Each imported case is paired with a control that does the same work without crossing an import boundary, and every case checks its executed plan before it is timed. Best time on an M2:

The controls stay flat, which is what places the cost in validation rather than anywhere else.

Those numbers come from validating with std::str::from_utf8. It has a fast ASCII path but degrades to a byte at a time loop on non-ASCII text. Timing the validator alone on the same string shapes, 8192 rows per batch:

111 ns/row accounts for the whole multibyte regression above, and the boundary check costs under 1 ns/row. The last commit therefore switches the fast path to simdutf8, which is 7x faster on multibyte data and 3x on ASCII. Substituting 15.7 ns/row back into the cases above projects roughly +15% for the multibyte converted scan and +29% for multibyte native columnar to row, with ASCII near noise.

The simdutf8 numbers are measured at the validator, not end to end. Repeated attempts to re-time the full benchmark against the simdutf8 build on this machine were spoiled by background load (the same case varied up to 4x between runs), so the end to end table above is still the std build. A quiet machine or the EC2 micro benchmark runner (benchmarks/micro/run.py) should confirm the projection before this is taken as final.

Skipping validation where the producer is known to be native is not available as a shortcut: native columnar to row can receive strings built on the JVM, for example through a union, the in-memory cache, or broadcast.

How are these changes tested?

• Rust unit tests for the walker covering: valid input returned zero copy (pointer identical buffers); invalid bytes decoding to U+FFFD matching the JVM; the split codepoint boundary case; nested Dictionary / Struct / List / FixedSizeList / Map (decode and zero copy paths); null preservation; sliced arrays with a non zero starting offset; and trailing empty strings.
• A Rust test at the ScanExec import site proving an invalid UTF-8 column is decoded through the production per column path.
• Both suites pass against the simdutf8 fast path.
• A criterion benchmark for the valid fast path, and CometStringFfiImportBenchmark for the two import sites end to end.